#include "NodeImpl.h"
#include "Common.h"
#include "pb_base.pb.h"
#ifdef WS_SSL
#include "root_certificates.hpp"
#endif
namespace wsbus {


	NodeImpl::NodeImpl(int appId, const std::string &host, const std::string &port)
		: appType_(appId)
		, sessionId_(0)
		, host_(host)
		, port_(port)
		, status_(csOffline)
		, requestHandler_(std::bind(&NodeImpl::replyMessage, this, std::placeholders::_1, std::placeholders::_2))
		, reconnectCount_(0)
		, heartbeatSeconds_(10)
		, requestThreadTool_(1)
		, connector_(host, port)
	{
		ConnectorImpl::HandleCallback cb;
		cb.handleRead = std::bind(&NodeImpl::onRead, this, std::placeholders::_1);
		cb.handleConnected = std::bind(&NodeImpl::onConnectorConnected, this, std::placeholders::_1);
		cb.handleError = std::bind(&NodeImpl::onConnectorError, this, std::placeholders::_1);
		connector_.setHandleCallback(cb);
	}

	NodeImpl::~NodeImpl()
	{
	}

	void NodeImpl::setAuth(const std::string &username, const std::string &password)
	{
		username_ = username;
		password_ = password;
	}

	void NodeImpl::setHeartbeatInterval(int second)
	{
		heartbeatSeconds_ = second;
	}

	void NodeImpl::start()
	{
		connector_.connect();
	}

	PackagePtr NodeImpl::newRequestPackage(const MessagePtr &msg, int toAppType, int toSessionId)
	{
		Package::Address addr(appType_, sessionId_, toAppType, toSessionId);
		return Package::newRequestPackage(addr, msg);
	}

	PackagePtr NodeImpl::newRequestPackage(const std::string &command, const std::string &data, int toAppType, int toSessionId)
	{
		Package::Address addr(appType_, sessionId_, toAppType, toSessionId);
		return Package::newRequestPackage(addr, command, data);
	}

	int NodeImpl::sendMessage(const PackagePtr &reqMsg, PackagePtr &rspMsg, int msTimeout)
	{
		if (!status_) {
			return neNoConnect;
		}

		int result = neSuccess;
		bool waiting = true;
		postMessage(reqMsg, [&](int errCode, const PackagePtr &msg) {
			std::unique_lock<std::mutex> lock(syncMsgMutex_);
			result = errCode;
			waiting = false;
			rspMsg = msg;
			syncMsgCond_.notify_all();
		}, msTimeout);

		std::unique_lock<std::mutex> lock(syncMsgMutex_);
		auto now = std::chrono::system_clock::now();
		using namespace std::chrono_literals;
		if (!syncMsgCond_.wait_until(lock, now + msTimeout * 1ms, [&]() { return waiting == false; })) {
			ResponseHandler::Info info;
			responseHandler_.pick(reqMsg->header().msgId, info);
			return neSendTimeout;
		}

		if (result != neSuccess) {
			return result;
		}

		return rspMsg ? neSuccess : neUnkown;
	}

	int NodeImpl::postMessage(const PackagePtr &msg, const HandleMessage &cb, int msTimeout)
	{
		if (!status_) {
			return neNoConnect;
		}

		responseHandler_.put(msg, cb, msTimeout);
		postPackage(msg);
		return neSuccess;
	}

	int NodeImpl::pushMessage(const MessagePtr &msg)
	{
		if (!status_) {
			return neNoConnect;
		}
		
		Package::Address addr(appType_, sessionId_, 0, 0);
		auto package = Package::newPushPackage(addr, msg);
		postPackage(package);
		return neSuccess;
	}

	int NodeImpl::replyMessage(const PackagePtr &reqPackage, const MoreMessagePtr &rspMsg)
	{
		const auto &h = reqPackage->header();
		auto rspPackage = Package::newResponsePackage(Package::Address(appType_, sessionId_, h.fromAppType, h.fromSessionId), h.msgId, rspMsg);
		postPackage(rspPackage);
		return 0;
	}

	void NodeImpl::addRequestHandler(const std::string &cmd, const std::function<MoreMessagePtr(const PackagePtr&)> &cb)
	{
		requestHandler_.addHandle(cmd, cb);
	}

	void NodeImpl::addPublishHandler(const std::string &cmd, const std::function<void(const PackagePtr)> &cb)
	{
		publishHandler_.addHandle(cmd, cb);
	}

	void NodeImpl::setStatusChangedHandler(const std::function<void(int status)> &cb)
	{
		statusChangedHandler_ = cb;
	}


	void NodeImpl::postPackage(const PackagePtr &msg)
	{
		BufferPtr buffer = std::make_shared<beast::flat_buffer>();
		int size = Package::encode(msg, *buffer.get());
		if (size > 0) {
			buffer->commit(size);
			connector_.post(buffer);
		}
	}

	void NodeImpl::setStatus(ConnStatus status)
	{
		bool changed = (status_ != status);
		status_ = status;
		if (changed && statusChangedHandler_) {
			statusChangedHandler_(status);
		}
	}

	void NodeImpl::onConnect()
	{
		login();
		startHeartbeatTimer();
	}

	void NodeImpl::onRead(const BufferPtr &buffer)
	{
		if (!buffer) {
			return;
		}

		auto package = Package::decode(*buffer.get());
		if (package) {
			if (package->header().msgType == responseHandler_.type()) {
				responseHandler_.onHandle(package);
			}
			else if (package->header().msgType == requestHandler_.type()) {
				requestThreadTool_.enqueue(std::bind(&RequestHandler::onHandle, requestHandler_, package));
			}
			else if (package->header().msgType == publishHandler_.type()) {
				publishHandler_.onHandle(package);
			}
			else {
				LOG(WARNING) << "onRead unkown type:" << package->header().msgType;
			}
		}
	}

	void NodeImpl::onConnectorConnected(bool connected)
	{
		setStatus(connected ? csOnline : csOffline);
		if (connected) {
			reconnectCount_ = 0;
			onConnect();
		}
		LOG(INFO) << "connected:" << connected;
	}

	void NodeImpl::onConnectorError(int error)
	{
		if (error == 0) {
			return;
		}

		setStatus(csOffline);
		int wait = reconnectCount_ * 3;
		wait = std::max(3, wait);
		wait = std::min(wait, 60);
		LOG(WARNING) << "connector error:" << error << ", reconnect count:" << reconnectCount_ << ", wait:" << wait << "s";
		sleepSecond(wait);
		connector_.reconnect();
	}

	void NodeImpl::login()
	{
		auto loginReq = std::make_shared<ProtoBase::LoginReq>();
		loginReq->set_appid(appType_);
		loginReq->set_oldsessionid(sessionId_);
		loginReq->set_username(username_);
		loginReq->set_password(password_);

		auto subscribe = loginReq->mutable_subscribe();
		subscribe->set_type(1);
		auto cmdList = publishHandler_.cmdList();
		for (const auto &cmd : cmdList) {
			subscribe->add_cmdlist(cmd);
		}

		auto request = Package::newRequestPackage(Package::Address(appType_, sessionId_, atBus, 0), loginReq);
		postMessage(request, [this](int errCode, const PackagePtr &rsp) {
			if (errCode == 0) {
				auto msg = rsp->getMessage();
				auto loginRsp = static_cast<ProtoBase::LoginRsp*>(msg.get());
				this->sessionId_ = loginRsp->sessionid();
				setStatus(csLogined);
				LOG(WARNING) << "login response, appid:" << appType_ << ", sessionid:" << sessionId_;
			}
			else {
				setStatus(csLogouted);
				LOG(WARNING) << "login failed, errCode:" << errCode;
			}
		});
	}

	void NodeImpl::doHeartbeat(const boost::system::error_code &error)
	{
		if (error) {
			LOG(ERROR) << "doHeartbeat error:" << error.message();
			return;
		}
		if (heartbeatSeconds_ <= 0) {
			return;
		}
		
		auto restartTimer = [&]() {
			heartbeatTimer_->expires_at(heartbeatTimer_->expires_at() + boost::posix_time::seconds(3));
			heartbeatTimer_->async_wait(std::bind(&NodeImpl::doHeartbeat, this, std::placeholders::_1));
		};

		time_t t = currentMillisecond();
		if (t - connector_.lastWriteMessageTime() > heartbeatSeconds_ * 1000) {
			auto heartbeatReq = std::make_shared<ProtoBase::HeartbeatReq>();
			heartbeatReq->set_time(t);
			postMessage(Package::newRequestPackage(Package::Address(appType_, sessionId_, atBus, 0), heartbeatReq), [&](int errorCode, const PackagePtr &rspPackage) {
				if (errorCode) {
					LOG(ERROR) << "heartbeast error:" << errorCode;
				}
			});
		}
		restartTimer();
	}

	void NodeImpl::startHeartbeatTimer()
	{
		//heartbeatTimer_.reset(new boost::asio::deadline_timer(ioc_, boost::posix_time::seconds(3)));
		//heartbeatTimer_->async_wait(std::bind(&NodeImpl::doHeartbeat, this, std::placeholders::_1));
	}
}
